package com.wujian.love.rocketmq.produce;

import com.wujian.love.contants.RocketMQContants;
import com.wujian.love.service.impl.TransactionListenerImpl;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * @ClassName: TransactionProduce
 * @description:
 * @author: wujian
 * @Date: 2020-06-11 16:07
 **/
@Component
public class TransactionProduce implements InitializingBean {





    //生产组
    private static final String PRODUCE_GROUP_NAME ="Love_Test_Order_GroupName";


    private  TransactionMQProducer transactionProduce;


    private ExecutorService executorService;

    @Autowired
    private TransactionListenerImpl transactionListenerImpl;


    //初始化produce
    private TransactionProduce(){
        System.out.println("初始化Produce");
        this.transactionProduce = new TransactionMQProducer(PRODUCE_GROUP_NAME);
        this.executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName(PRODUCE_GROUP_NAME+"-check-thread");
                System.out.println(PRODUCE_GROUP_NAME+"-check-thread");
                return thread;
            }
        });
        this.transactionProduce.setExecutorService(executorService);
        this.transactionProduce.setNamesrvAddr(RocketMQContants.NAMESERVER);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.transactionProduce.setTransactionListener(transactionListenerImpl);
        start();
    }

    private void start(){
        try {
            this.transactionProduce.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    private void shutDown(){
        this.transactionProduce.shutdown();

    }

    public TransactionSendResult sendMessage(Message message, Object argument){
        TransactionSendResult result= null;
        try {
            result = this.transactionProduce.sendMessageInTransaction(message, argument);

        }catch (Exception e){
            e.printStackTrace();
        }
        return  result;
    }


    /**
     * @Author WuJian
     * @Description 发送订单延时消息
     * @Date 10:21 2020/7/3
     * @Param [message(订单消息), argument]
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     **/
    public TransactionSendResult sendCancelOrderMessage(Message message, Object argument){
        TransactionSendResult result= null;
        try {
            result = this.transactionProduce.sendMessageInTransaction(message, argument);

        }catch (Exception e){
            e.printStackTrace();
        }
        return  result;
    }
}
